package cn.tpshion.pulsar;

import cn.tpshion.pulsar.config.PulsarConsumerConfiguration;
import cn.tpshion.pulsar.config.PulsarProducerConfiguration;
import cn.tpshion.pulsar.listener.DefaultMessageListener;
import cn.tpshion.pulsar.listener.support.PulsarConsumerHandle;
import cn.tpshion.pulsar.support.PulsarProperties;
import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.shade.org.apache.commons.lang3.StringUtils;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;

@Configuration
@ConditionalOnClass(PulsarClient.class)
@EnableConfigurationProperties(PulsarProperties.class)
@Import({PulsarConsumerHandle.class, DefaultMessageListener.class})
public class PulsarAutoConfiguration {

    @Bean(destroyMethod = "close")
    public PulsarClient pulsarClient(PulsarProperties properties) throws PulsarClientException {
        ClientBuilder builder = PulsarClient.builder();
        builder.serviceUrl(properties.getServiceUrl());
        if(StringUtils.isNoneBlank(properties.getToken())){
            builder.authentication(AuthenticationFactory.token(properties.getToken()));
        }
        builder.operationTimeout(properties.getOperationTimeout().getTimeOut(), properties.getOperationTimeout().getTimeUnit());
        builder.ioThreads(properties.getIoThreads());
        builder.listenerThreads(properties.getListenerThreads());
        builder.connectionMaxIdleSeconds(properties.getConnectionMaxIdleSeconds());
        builder.enableTcpNoDelay(properties.isEnableTcpNoDelay());
        builder.keepAliveInterval(properties.getKeepAliveInterval().getTimeOut(), properties.getKeepAliveInterval().getTimeUnit());
        builder.connectionTimeout(properties.getConnectionTimeout().getTimeOut(), properties.getConnectionTimeout().getTimeUnit());
        return builder.build();
    }

    @Bean
    @ConditionalOnMissingBean(PulsarProducerConfiguration.class)
    public PulsarProducerConfiguration pulsarProducerConfiguration(PulsarProperties properties, PulsarClient pulsarClient){
        return new PulsarProducerConfiguration(properties, pulsarClient);
    }

    @Bean
    @ConditionalOnMissingBean(PulsarConsumerConfiguration.class)
    public PulsarConsumerConfiguration pulsarConsumerConfiguration(PulsarProperties properties, PulsarClient pulsarClient){
        return new PulsarConsumerConfiguration(properties, pulsarClient);
    }

}
